package org.zhadoop.streaming.application;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zhadoop.streaming.exception.ErrorCode;
import org.zhadoop.streaming.exception.StreamingException;
import org.zhadoop.streaming.operator.IRichOperator;
import org.zhadoop.streaming.operator.OutputOperator;
import org.zhadoop.streaming.operator.functionstream.SplitOP;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;

/**
 * 应用中的算子管理
 * @author zwd
 *
 */
public class OperatorMng {
	private static final Logger LOG = LoggerFactory.getLogger(OperatorMng.class);
	
	//DFG排序后的功能算子列表，作为创建Storm拓扑顺序的基础
	private List<IRichOperator> sortedFunctions;
	
	private Map<String,IRichOperator> inputs;
	
	private Map<String,IRichOperator> functions;
	
	private Map<String,IRichOperator> outputs;
	
	public OperatorMng() {
		this.inputs = Maps.newLinkedHashMap();
		this.outputs = Maps.newLinkedHashMap();
		this.functions = Maps.newLinkedHashMap();
		this.sortedFunctions = Lists.newArrayList();
	}
	
	/**
	 * 算子名称是否有效：如果不重复则有效
	 * @param opName
	 * @return
	 */
	public boolean isNameValid(String opName){
		LOG.debug("isNameValid enter, the opName is:{}.", opName);
		if(!inputs.containsKey(opName) && !functions.containsKey(opName)){
			return true;
		}
		return false;
	}
	
	/**
	 * 添加源算子信息
	 * @param source
	 * @return
	 */
	public boolean addInputStreamOperator(IRichOperator source){
		if(isNameValid(source.getOperatorId())){
			inputs.put(source.getOperatorId(), source);
			return true;
		}
		return false;
	}
	
	
	/**
	 * 添加输出算子
	 * @param output
	 * @return
	 */
	public boolean addOutputStreamOperator(IRichOperator output){
		if(isNameValid(output.getOperatorId())){
			outputs.put(output.getOperatorId(), output);
			return true;
		}
		return false;
	}
	
	/**
	 * 添加功能算子
	 * @param operator
	 * @return
	 */
	public boolean addFunctionStreamOperator(IRichOperator operator){
		if(isNameValid(operator.getOperatorId())){
			functions.put(operator.getOperatorId(), operator);
			return true;
		}
		return false;
	}
	
	/**
	 * 获取输出算子
	 * @return
	 */
	public List<IRichOperator> getOutputOps(){
		List<IRichOperator> s = new ArrayList<>();
		for(Entry<String,IRichOperator> m : outputs.entrySet()){
			s.add(m.getValue());
		}
		return s;
	}
	
	/**
	 * 获的所有功能算子信息
	 * @return
	 */
	public List<IRichOperator> getFunctionOps(){
		LOG.debug("GetFunctionOps enter.");
		List<IRichOperator> s= new ArrayList<>();
		for(Entry<String, IRichOperator> m : functions.entrySet()){
			s.add(m.getValue());
		}
		return s;
	}
	
	/**
	 * 获取所有的源算子信息
	 * @return
	 */
	public List<IRichOperator> getSourceOps(){
		List<IRichOperator> s = new ArrayList<>();
		for(Entry<String, IRichOperator> m : inputs.entrySet()){
			s.add(m.getValue());
		}
		return s;
	}
	
	/**
	 * 产生有向无环图DAG
	 * 
	 * 对有向图进行拓扑排序
	 * 
	 * 无后继的顶点优先拓扑排序方法
	 * 1、思想方法
	 * 该方法的每一步均是输出当前无后继（即出度为0)的顶点。
	 * 对于一个DAG，按此方法的输出的序列是逆拓扑次序。
	 * 因此设计一个栈（或向量）T来保存输出的顶点序列，即可得拓扑序列
	 * 若T是栈，则每当输出顶点时，只需要做入栈的操作，排序完成时将栈中的顶点依次出栈即可得拓扑序列。
	 * 若T是向量，则将输出的顶点从T[n-1]开始依次从后面往前放，即可保证T中存储的顶点是拓扑序列
	 * 
	 * 生成功能算子拓扑顺序
	 * @return
	 */
	public List<IRichOperator> getFunctionOpsOrder() throws StreamingException{
		LOG.info("GenFunctionOpsOrder enter.");
		validateOperators();
		
		//所有的当前已删除的输出流的名称
		List<String> deletedStreamName = Lists.newArrayList();
		
		//获得所有的Soource 算子的输出流名称
		for(Entry<String,IRichOperator> et : inputs.entrySet()){
			deletedStreamName.add(et.getValue().getOutputStream());
		}
		
		if(deletedStreamName.size() == 0){
			LOG.debug("All the source operators have no output stream.");
            return null;
		}
		
		//获得当前所有未排序的功能算子名称
		Map<String,IRichOperator> functionsAndOutputs = functionAndOutputOperators();
		
		Set<String> unSortedFunctions = Sets.newHashSet();
		unSortedFunctions.addAll(functionsAndOutputs.keySet());
		
		IRichOperator fun = null;
		boolean qualified = true;
		int presortedSize = 0;
		boolean stopFlag = false;
		
		//在未排序的功能算子中遍历处理
		while(unSortedFunctions.size() > 0 && !stopFlag){
			presortedSize = sortedFunctions.size();
			Iterator<String> iterator = unSortedFunctions.iterator();
			
			while(iterator.hasNext()){
				String funname = iterator.next();
				fun = functionsAndOutputs.get(funname);
				List<String> inputStreams = fun.getInputStream();
				if(inputStreams == null){
					//为空，则没有输入算子，异常，退出
					break;
				}
				
				for(String inputname : inputStreams){
					if(!deletedStreamName.contains(inputname)){
						qualified = false;
						break;
					}
				}
				
				/*
				 *如果该算子的所有输入流名称均出现在已删除名称中，则
				 *将该算子加入到排序算子的列表中 
				 *并将其输出流名称加入到已删除流名称
				 *在未排序算子名称中删除该算子的名称
				 */
				if(qualified){
					sortedFunctions.add(fun){
						if(fun instanceof OutputOperator){
							/**
							 * split 算子有多个输出
							 * 这个只是暂时修改，目前的接口还不支持多流的输出
							 */
							if(fun instanceof SplitOP){
								SplitOP sop = (SplitOP)fun;
								
							}
							
						}
					}
				}
				
			}
			
			
			
			
		}
				
				
	}

	
	private Map<String, IRichOperator> functionAndOutputOperators() {
		Map<String,IRichOperator> map = Maps.newHashMap();
		map.putAll(functions);
		map.putAll(outputs);
		return map;
	}

	private void validateOperators() throws StreamingException{
		if(null == sortedFunctions || sortedFunctions.size() > 0){
			StreamingException exception = new StreamingException(ErrorCode.PLATFORM_INVALID_TOPOLOGY);
            LOG.error(ErrorCode.PLATFORM_INVALID_TOPOLOGY.getFullMessage(), exception);
            throw exception;
		}
	}
	
	
}
